package com.qezhhnjy.flink.source;

import cn.hutool.core.thread.ThreadUtil;
import com.qezhhnjy.flink.bean.SensorReading;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * @author zhaoyangfu
 * @date 2021/8/27-13:41
 */
public class MySensor implements SourceFunction<SensorReading> {

    private boolean running = true;

    @Override
    public void run(SourceContext<SensorReading> ctx) throws Exception {
        Random random = new Random();
        while (running) {
            for (int i = 1; i <= 10; i++) {
                ctx.collect(new SensorReading("sensor_" + i,
                        System.currentTimeMillis(), i * random.nextDouble() * 10));
            }
            ThreadUtil.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        this.running = false;
    }
}
